package mr

import (
	"fmt"
	"io/ioutil"
	"log"
	"strconv"
	"strings"
	"sync"
	"time"
)
import "net"
import "os"
import "net/rpc"
import "net/http"

type Master struct {
	// Your definitions here.
	tasks        []*Task
	rSourceFiles map[int][]string
	mux          sync.Mutex
	// 0: map 1: reduce
	phase int
}

type Task struct {
	taskId     int
	taskType   string
	filePath   string
	filePaths  []string
	done       bool
	processing bool
	startTime  time.Time
}

func (m *Master) getTask(id int) *Task {
	for _, task := range m.tasks {
		if task.taskId == id {
			return task
		}
	}
	return nil
}

// Your code here -- RPC handlers for the worker to call.

//
// an example RPC handler.
//
// the RPC argument and reply types are defined in rpc.go.
//
func (m *Master) Example(args *ExampleArgs, reply *ExampleReply) error {
	reply.Y = args.X + 1
	return nil
}

// Get a task for client call
func (m *Master) GetTask(args *GetTaskArgs, reply *GetTaskReply) error {
	m.mux.Lock()
	for _, task := range m.tasks {
		if task.processing || task.done {
			continue
		}
		reply.TaskId = task.taskId
		reply.FilePath = task.filePath
		reply.FilePaths = task.filePaths
		reply.DidGetTask = true
		reply.TaskType = task.taskType
		task.processing = true
		task.startTime = time.Now()
		break
	}
	m.mux.Unlock()
	return nil
}

// Finish a task for a client
func (m *Master) FinishTask(args *FinishTaskArgs, reply *FinishTaskReply) error {
	m.mux.Lock()
	defer m.mux.Unlock()
	task := m.getTask(args.TaskId)
	if task.done {
		return nil
	}
	task.processing = false
	task.done = true
	// store to reduce task
	for _, filename := range args.Filenames {
		n := strings.Split(filename, "-")[2]
		if v, err := strconv.Atoi(n); err == nil {
			m.rSourceFiles[v] = append(m.rSourceFiles[v], filename)
		}
	}
	log.Printf("%s task, taskId: %d is finished.\n", task.taskType, args.TaskId)
	// if map task done, change phase to "reduce"(1)
	if m.phase == 0 && m.isMapTaskDone() {
		m.generateReduceTask()
		m.phase = 1
		log.Printf("switch to phase 1, begin reduce...")
	}
	return nil
}

func (m *Master) isMapTaskDone() bool {
	done := true
	for _, v := range m.tasks {
		if v.taskType == "map" && (v.processing || !v.done) {
			done = false
		}
	}
	return done
}

func (m *Master) generateReduceTask() {
	m.tasks = []*Task{}
	for i := 0; i < 10; i++ {
		m.tasks = append(m.tasks, &Task{
			taskId:     i,
			taskType:   "reduce",
			filePaths:  m.rSourceFiles[i],
			processing: false,
			done:       false,
		})
	}
}

func (m *Master) combineReduceResultFiles() {
	outFile, err := os.Create("mr-out-result")
	if err != nil {
		log.Fatalf("can not create file: mr-out-1")
	}
	for i := 0; i < 10; i++ {
		task := m.tasks[i]
		outPath := fmt.Sprintf("mr-out-%v", i)
		file, err := os.Open(outPath)
		if err != nil {
			log.Fatalf("can not open file: %v", task.filePath)
		}
		content, err := ioutil.ReadAll(file)
		if err != nil {
			log.Fatalf("can not read file: %v", task.filePath)
		}
		outFile.Write(content)
		log.Printf("writed file: %v, len: %d", outPath, len(content))
		os.Remove(outPath)
	}
	outFile.Close()
	os.RemoveAll("out")
	log.Println("combine done.")
}

// scan to find task which is timeout
func (m *Master) scanTimeoutTask() {
	for {
		for _, task := range m.tasks {
			elapsed := time.Now().Unix() - task.startTime.Unix()
			if task.processing && elapsed >= 10 {
				task.processing = false
				task.done = false
				log.Printf("%v task [%v] is timeout!", task.taskType, task.taskId)
			}
		}
		time.Sleep(time.Millisecond * 800)
	}
}

//
// start a thread that listens for RPCs from worker.go
//
func (m *Master) server() {
	rpc.Register(m)
	rpc.HandleHTTP()
	//l, e := net.Listen("tcp", ":1234")
	sockname := masterSock()
	os.Remove(sockname)
	l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	go http.Serve(l, nil)
}

//
// main/mrmaster.go calls Done() periodically to find out
// if the entire job has finished.
//
func (m *Master) Done() bool {
	ret := true

	if m.phase == 0 {
		return false
	}

	if len(m.tasks) == 0 {
		return false
	}

	// Your code here.
	for _, task := range m.tasks {
		if !task.done {
			return false
		}
	}

	m.combineReduceResultFiles()
	log.Println("Tasks all done.")
	return ret
}

//
// create a Master.
// main/mrmaster.go calls this function.
// nReduce is the number of reduce mapTasks to use.
//
func MakeMaster(files []string, nReduce int) *Master {
	m := Master{}
	m.phase = 0
	m.tasks = make([]*Task, 0, len(files))
	log.Println(files)
	// Your code here.

	// create map tasks
	for i, filePath := range files {
		task := &Task{
			taskId:     i,
			taskType:   "map",
			filePath:   filePath,
			done:       false,
			processing: false,
		}
		m.tasks = append(m.tasks, task)
	}

	m.rSourceFiles = make(map[int][]string, nReduce)
	for i := 0; i < nReduce; i++ {
		m.rSourceFiles[i] = []string{}
	}

	m.server()

	// start to scan
	go m.scanTimeoutTask()
	return &m
}
